package com.atguigu.Flink.sql.query;

import com.atguigu.Flink.POJO.Event;
import com.atguigu.Flink.function.ClickSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Flink10_Deduplication {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> ds = env.addSource(new ClickSource());
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Schema schema = Schema.newBuilder()
                .column("user", "STRING")
                .column("url", "STRING")
                .column("ts", "BIGINT")
                //指定处理时间字段
                .columnByExpression("pt", "proctime()")
                //指定事件时间字段
                .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts, 3) ")
                .watermark("et", "et - INTERVAL '0' SECOND ")
                .build();
        Table table = tableEnv.fromDataStream(ds, schema);
        tableEnv.createTemporaryView("t1",table);
        //通过窗口处理，给每条数据补充窗口信息 开始和结束时间
        String windowsql =
                " select url ,  et , window_start , window_end " +
                        " from table (tumble(TABLE t1 , DESCRIPTOR(et) , INTERVAL '10' SECOND)) " ;
        Table windowTable = tableEnv.sqlQuery(windowsql);
        tableEnv.createTemporaryView("t2",windowTable);

        //给相同的窗口中的相同的url求排名
        String overSql =
                " select url ,et , window_start, window_end , " +
                        " row_number() over(partition by window_start, window_end , url order by et desc ) rk " +
                        " from  t2 " ;
        Table overTable = tableEnv.sqlQuery(overSql);
        tableEnv.createTemporaryView("t3" , overTable);

        //求top1
        String rankSql =
                " select url , et , window_start , window_end, rk " +
                        " from t3 " +
                        " where rk = 1 " ;
        tableEnv.sqlQuery(rankSql).execute().print();



        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
